package com.haiyou.dc.canal.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.haiyou.dc.canal.canal.handler.RowChangeHandler;
import com.haiyou.dc.canal.canal.handler.RowChangeRabbitHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.Assert;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;

@Slf4j
public class CanalClient {

    private String destination;
    private String subTables;
    final int batchSize = 1000;
    protected volatile boolean running = false;
    protected CanalConnector connector;
    //数据处理器
    protected List<RowChangeHandler> rowChangeHandlers = new ArrayList<>();
    protected Thread thread = null;
    protected static String row_format = "->binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {} , delay : {}ms";

    public CanalClient(CanalConnector connector, String destination, String subTables) {
        log.info("connector=" + connector + ",destination" + destination + ",subTables=" + subTables);
        this.connector = connector;
        this.destination = destination;
        this.subTables = subTables;
    }

    public void start() {
        Assert.notNull(connector, "connector is null");
        thread = new Thread(() -> {
            process();
        });
        thread.setUncaughtExceptionHandler((t, e) -> {
            log.error("parse events has an error", e);
        });
        thread.start();
        running = true;
    }

    protected void stop() {
        if (!running) {
            return;
        }
        running = false;
        if (thread != null) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                // ignore
            }
        }
    }

    protected void process() {
        try {
            connector.connect();
//            connector.subscribe("mustang_v2\\..*");
            connector.subscribe(subTables);
            while (running) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getEntries().size();
                //没有变化的数据
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                    }
                } else {
                    log.info("有数据变更====>" + message.getId() + ",size=" + size);
                    rowdataHandle(message.getEntries());
                }
                // 提交确认
                connector.ack(batchId);
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
        } catch (Exception e) {
            log.error("process error!", e);
        } finally {
            connector.disconnect();
        }
    }

    protected void rowdataHandle(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            long executeTime = entry.getHeader().getExecuteTime();
            long delayTime = System.currentTimeMillis() - executeTime;
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                CanalEntry.RowChange rowChange = null;
                try {
                    rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    log.error("parse events has an error, data:" + entry.toString(), e);
                    return;
                }
                log.info(row_format, new Object[]{entry.getHeader().getLogfileName(),
                        String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(),
                        entry.getHeader().getTableName(), rowChange.getEventType(),
                        String.valueOf(entry.getHeader().getExecuteTime()), String.valueOf(delayTime)});
                //获取变化的数据，给具体的业务处理器处理
                for (RowChangeHandler h : rowChangeHandlers) {
                    h.handler(entry, rowChange);
                }
            }
        }
    }

    public void registerRowDataHandler(RowChangeHandler rowChangeHandler) {
        rowChangeHandlers.add(rowChangeHandler);
    }

    public static void main(String[] args) throws InterruptedException {
        log.info("binlog server start......");
        String subTables = RowChangeRabbitHandler.tables;
        String ip = "192.168.1.105";
        String destination = "example";
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11111), destination, "", "");
        CanalClient canalClient = new CanalClient(connector, destination, subTables);
        canalClient.registerRowDataHandler(new RowChangeRabbitHandler());
        canalClient.start();


        Thread.sleep(1000000);
    }
}
